package com.aliyun.adb.contest;

import com.aliyun.adb.contest.spi.AnalyticDB;

import java.io.*;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.*;

public class SimpleAnalyticDB implements AnalyticDB {

    private final Map<String, List<Long>> data = new HashMap<String, List<Long>>();

    private static String WORK_DIR;

    public static void main(String[] args) throws Exception {
      //  System.out.println("===========os.name:"+System.getProperties().getProperty("os.name"));
        File fileName = new File("./my_test_data/2output1");
        BufferedReader reader = new BufferedReader(new FileReader(fileName),(int)Const.BUFFER_SIZE); // 如果是读大文件，设置缓存
        ArrayList<Long> list = new ArrayList<>();

        String rawRow;
        while ((rawRow = reader.readLine()) != null) {
            list.add(Long.parseLong(rawRow));
        }
        Collections.sort(list);
        for(int i = 0;i < list.size();i++){
            if(list.get(i) == 1470939429013645136L){
                System.out.println(i);
            }
        }
    }

    /**
     *
     * The implementation must contain a public no-argument constructor.
     *
     */
    public SimpleAnalyticDB() {
    }

    @Override
    public void load(String tpchDataFileDir, String workspaceDir) throws Exception {
        File dir = new File(tpchDataFileDir);
        this.WORK_DIR = workspaceDir;
        for (File dataFile : dir.listFiles()) {
            System.out.println("Start loading table " + dataFile.getName()+" table.size:" + dataFile.length());
            SplitBigFileTools splitBigFileTools = new SplitBigFileTools();
            splitBigFileTools.readV3(dataFile, workspaceDir);
        }
    }


    //here can use offset
    @Override
    public String quantile(String table, String column, double percentile) throws Exception {
        System.out.println("conlumn is: "+column + "  percentile is: "+percentile);
        int rank = (int) Math.round((Const.LINE_NUM) * percentile);
        long t1 = System.currentTimeMillis();
        boolean isLeftSearch = true;
        if(column.equals("L_PARTKEY")) isLeftSearch = false;

        long[] sumOne = SplitBigFileTools.SUM_ONE;
        long[] sumTwo = SplitBigFileTools.SUM_TWO;
        long[] search = sumOne;
        if(!isLeftSearch){
            search = sumTwo;
        }
        int l = 0,r = search.length - 1;
        while(l < r){
            int mid = (l + r)/2;
            if(rank <= search[mid]) r = mid;
            else l = mid + 1;
        }

        if(l != 0){
            rank -= search[l-1];
        }

        String fileName =  WORK_DIR + "/" + l + "output1";
        if(!isLeftSearch){
               fileName = WORK_DIR + "/" + l + "output2";
        }
        System.out.println("search in file: " + fileName +"  and in line: " + rank);
        //here can change read io


        FileChannel fc = new RandomAccessFile(fileName, "r").getChannel();
        MappedByteBuffer bf = fc.map(FileChannel.MapMode.READ_ONLY, 0 ,fc.size());

       /* BufferedReader reader = new BufferedReader(new FileReader(fileName),(int)Const.BUFFER_SIZE); // 如果是读大文件，设置缓存
        String rawRow;
        while ((rawRow = reader.readLine()) != null) {
            list.add(Long.parseLong(rawRow));
        }
        Collections.sort(list);
        long res = list.get(rank-1);*/

        long res = handleQuery(bf, rank);
        System.out.println("search delay: "+(System.currentTimeMillis()-t1));

        if(Const.USE_OFFSET && l != 0){
            res += Const.COM_MAP[l-1];
        }
        //return "";
        return res+"";
    }

    public long handleQuery(MappedByteBuffer bf,int rank){
        List<Long> list = new ArrayList<>();
        long t = 0;
        while(bf.hasRemaining()){
            byte b = bf.get();
            if(b == '\n'){
                list.add(t);
                t = 0;
            } else{
                if(b >= 48 && b <= 57){
                    t = t*10 + (b-48);
                }
            }
        }
        Collections.sort(list);
       // System.out.println(list.size() +"  " + list);
        return list.get(rank-1);
    }

}
